package cn.com.fattiger.framework.mec.controller;

import cn.com.fattiger.framework.mec.config.CustomConfig;
import cn.com.fattiger.framework.mec.model.ZooNode;
import cn.com.fattiger.framework.mec.model.ZooResponse;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.ModelMap;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.net.URLDecoder;
import java.util.*;


@Controller
@RequestMapping("zookeeper")
public class ZookeeperController extends BaseController{

	private Map<String,ZooKeeper> ZK_MAP = new HashMap<String,ZooKeeper>();
	
	@Autowired
	private CustomConfig customConfig;
	
	@PostConstruct
	public void init() throws Exception {
		log.info("init zookeeper config.");
		String zookeepers = customConfig.getZookeepers();
		if(StringUtils.isNotEmpty(zookeepers)){
			String[] zookeeperArr = zookeepers.split(",");
			for(String zk:zookeeperArr){
                ZooKeeper zooKeeper  = new ZooKeeper(zk, customConfig.getZookeepersTimeout(), new ZkWatcher(zk,customConfig.getZookeepersTimeout()));
                ZK_MAP.put(zk, zooKeeper);
            }
		}
	}
	@RequestMapping("list.do")
	public String list(
            @RequestParam(value = "parent",defaultValue = "/",required = false) String parent,
            @RequestParam(value = "zkConnect",required = false) String zkConnect,
            ModelMap map){
		List<String> zkList = new ArrayList<String>();
		for(String s:ZK_MAP.keySet()){
			zkList.add(s);
		}
		map.put("zkList", zkList);
        ZooResponse response = new ZooResponse();
        if(zkConnect == null){
            zkConnect = zkList.get(0);
        }
        try{
            ZooKeeper zooKeeper  = ZK_MAP.get(zkConnect);
            if(zooKeeper.getState().isAlive()){
                List<String> children = zooKeeper.getChildren(parent,false);
                List<ZooNode> nodes = new ArrayList<ZooNode>();
                for(String child:children){
                    ZooNode node = new ZooNode();
                    node.setName(child);
                    node.setDecodeName(URLDecoder.decode(child, "UTF-8"));
                    node.setParent(parent);
                    node.setNodeStat(zooKeeper.exists((parent.equals("/")?"":parent)+"/"+child,false));
                    if(zooKeeper.getChildren((parent.equals("/")?"":parent)+"/"+child,false).size()>0){
                        node.setNodeList(new ArrayList<ZooNode>());
                    }else{
                        node.setChildNodes(new ArrayList<ZooNode>());
                    }
                    nodes.add(node);
                }
                Collections.sort(nodes);
                response.setNodeList(nodes);
            }else{
                response.setState(ZooResponse.State.REMOTE_ERROR);
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error("failed to load node {} children",e);
            response.setState(ZooResponse.State.FAILED);
        }
        map.put("result", response);
        map.put("zkConnect", zkConnect);
	    return "zookeeper/zookeeper_list";
	}

	private class ZkWatcher implements Watcher{
        private String host;
        private Integer timeOut;
        public ZkWatcher(String host, Integer timeOut) {
            this.host = host;
            this.timeOut = timeOut;
        }

        @Override
        public void process(WatchedEvent watchedEvent) {
            if(watchedEvent.getState()== Event.KeeperState.Expired||watchedEvent.getState()==Event.KeeperState.Disconnected){
                try {
                    ZooKeeper zooKeeper = ZK_MAP.get(host);
                    if(zooKeeper!=null){
                        try {
                            zooKeeper.close();
                        } catch (InterruptedException e) {
                            //do nothing
                        }
                    }
                    ZK_MAP.put(host,new ZooKeeper(host,timeOut,this));
                } catch (IOException e) {
                    log.error("failed to reconnect zookeeper server",e);
                }
            }
        }
    }
}
